import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time

object TimeWindowTest {
  def main(args: Array[String]): Unit = {

    val env = StreamExecutionEnvironment.getExecutionEnvironment
    val stream1 = env.readTextFile("D:\\codes\\IDEA_prjects\\FlinkDemo\\data\\sensor.txt")
    val stream2 = stream1.map(
      data => {
        val strings = data.split(",")
        TempData(strings(0), strings(1).toLong, strings(2).toDouble)
      }
    )

    //TODO 滚动窗口
    val minTempPerWindow = stream2
      .map(r => (r.id, r.temp))
      .keyBy(_._1)
      .timeWindow(Time.seconds(15))
      // x._2.min(y._2) 取x, y的_.2的最小值
      .reduce((x, y) => (x._1, x._2.min(y._2)))

    //TODO 滑动窗口
    val minTempPerWindow2 = stream2
      .map(r => (r.id, r.temp))
      .keyBy(_._1)
      .timeWindow(Time.seconds(15), Time.seconds(5))
      .reduce((x, y) => (x._1, x._2.min(y._2)))

    env.execute()
  }
  case class TempData(id: String, tel: Long, temp: Double)

}
